package model

import (
	"census/common/records"
	"census/common/utils"
	"census/context"
	"encoding/json"
	"fmt"
	"github.com/golib/helper"
	"github.com/golib/logger"
	"github.com/golib/mysql"
	"sync"
	"time"
)

//任务池

type Pool struct {
	appName string
	mutex   *sync.Mutex
	//任务信道
	TaskChan map[int]chan struct{}
	//挂起
	BufferTaskList map[int][]*Task
	//正在进行
	ProceedTaskList map[int]map[string]*Task
	//失败
	FailTaskList map[int][]*Task
	SuccessCount int
	FailCount    int
	RunningCount int
}

func NewTaskPool(appName string) *Pool {
	var p = new(Pool)
	p.appName = appName
	p.mutex = new(sync.Mutex)
	p.TaskChan = make(map[int]chan struct{})
	p.BufferTaskList = make(map[int][]*Task)
	p.ProceedTaskList = make(map[int]map[string]*Task)
	p.FailTaskList = make(map[int][]*Task)
	return p
}

func (p *Pool) ParallelTaskChan(taskId int) chan struct{} {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	if _, ok := p.TaskChan[taskId]; !ok {
		num := context.GetTask(taskId).ThreadNum
		if num < 1 {
			num = 1
		}
		p.TaskChan[taskId] = make(chan struct{}, num)
	}
	return p.TaskChan[taskId]
}

func (p *Pool) DispatchTask(task *Task) {
	select {
	case p.ParallelTaskChan(task.taskId) <- struct{}{}:
		go task.Run()
	default:
		p.EnQueueBufferTask(task)
	}
}

func (p *Pool) EnQueueBufferTask(task *Task) {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	if _, ok := p.BufferTaskList[task.taskId]; !ok {
		p.BufferTaskList[task.taskId] = make([]*Task, 0)
	}
	p.BufferTaskList[task.taskId] = append(p.BufferTaskList[task.taskId], task)
	task.Buffer()
}

func (p *Pool) DeQueueBufferTask(taskId int) *Task {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	if _, ok := p.BufferTaskList[taskId]; !ok {
		return nil
	}
	var taskList = p.BufferTaskList[taskId]
	if len(taskList) == 0 {
		return nil
	}
	var task = taskList[0]
	if len(taskList) == 1 {
		p.BufferTaskList[taskId] = make([]*Task, 0)
	} else {
		p.BufferTaskList[taskId] = taskList[1:]
	}
	return task
}

func (p *Pool) EnQueueFailTask(task *Task) {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	if _, ok := p.FailTaskList[task.taskId]; !ok {
		p.FailTaskList[task.taskId] = make([]*Task, 0)
	}
	p.FailCount++
	p.FailTaskList[task.taskId] = append(p.FailTaskList[task.taskId], task)
}

func (p *Pool) AddProceedTask(task *Task) {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	if _, ok := p.ProceedTaskList[task.taskId]; !ok {
		p.ProceedTaskList[task.taskId] = make(map[string]*Task, 0)
	}
	p.RunningCount++
	p.ProceedTaskList[task.taskId][task.GetTaskUUId()] = task
}

func (p *Pool) RemoveProceedTask(task *Task) {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	if _, ok := p.ProceedTaskList[task.taskId]; ok {
		p.RunningCount--
		delete(p.ProceedTaskList[task.taskId], task.GetTaskUUId())
	}
}

func (p *Pool) Monitor(ch *Channel) {
	for {
		select {
		case <-GetProChan():
			logger.Info("processor-stop", utils.MakeUUID(), "receive all processor stop signal. now stop task")
			p.stopAllTask(ch)
			return
		case task := <-ch.TaskProceed:
			p.AddProceedTask(task)
		case task := <-ch.TaskFail:
			logger.Warning("task-end", task.GetTaskUUId(), "task:[fail]", "task data:", task.GetTaskData())
			p.EnQueueFailTask(task)
		case task := <-ch.TaskSuccess:
			logger.Info("task-end", task.GetTaskUUId(), "task:[success]", "task data:", task.GetTaskData())
			p.SuccessCount++
		case task := <-ch.TaskEnd:
			p.RemoveProceedTask(task)
			<-p.ParallelTaskChan(task.taskId)
			var t = p.DeQueueBufferTask(task.taskId)
			if t != nil {
				p.DispatchTask(t)
			}
		}
	}
}

func (p *Pool) stopAllTask(ch *Channel) {
ForBegin:
	for {
		select {
		case task := <-ch.TaskProceed:
			p.AddProceedTask(task)
		case task := <-ch.TaskFail:
			logger.Warning("task-end", task.GetTaskUUId(), "task:[fail]", "task data:", task.GetTaskData())
			p.EnQueueFailTask(task)
		case task := <-ch.TaskSuccess:
			logger.Info("task-end", task.GetTaskUUId(), "task:[success]", "task data:", task.GetTaskData())
			p.SuccessCount++
		case task := <-ch.TaskEnd:
			p.RemoveProceedTask(task)
			<-p.ParallelTaskChan(task.taskId)
		default:
			if p.RunningCount == 0 {
				//concurrency problems process
				processor, err := GetProcessor(p.appName)
				if err != nil {
					goto ForEnd
				}
				select {
				case <-processor.GetSem():
					time.Sleep(2 * time.Second)
					goto ForBegin
				default:
					goto ForEnd
				}
			}
		}
	}
ForEnd:
	//显示在终端
	p.terminal()
	//所有task stop
	ch.AllTaskStop <- true
}

//任务恢复
func (p *Pool) Recover(appName string, ch *Channel) {
	var db = mysql.GetInstance(false)
	var taskListRecord = db.Find(records.RecordNameCensusTaskInfo).Select("id", "uuid", "action", "data").
		Where("app_name", "=", appName).AndWhere("status", "=", TaskStatusBuffer).
		Execute().FetchAll()
	if taskListRecord.Len() == 0 {
		return
	}
	var taskIdMap = make(map[int]bool)
	for _, o := range taskListRecord.AllRecord() {
		r := o.(*records.CensusTaskInfo)
		t := new(Task)
		t.id = r.Id
		t.uuid = r.Uuid
		t.taskId = r.Action
		t.appName = appName
		t.taskFunc = context.GetTaskModuleMap(t.taskId)
		var data interface{}
		if r.Data != "" {
			err := json.Unmarshal([]byte(r.Data), &data)
			if err != nil {
				continue
			}
		}
		t.data = data
		p.EnQueueBufferTask(t)
		taskIdMap[t.taskId] = true
		fmt.Println("taskPriId", t.id, "taskUUId", t.uuid, "action", t.taskId, "appName", t.appName, "data", t.data)
	}
	//预置go-routine 数
	for taskId, _ := range taskIdMap {
		t := p.DeQueueBufferTask(taskId)
		p.DispatchTask(t)
		num := context.GetTask(taskId).ThreadNum
		for num > 1 {
			t := p.DeQueueBufferTask(taskId)
			if t == nil {
				break
			}
			p.DispatchTask(t)
			num--
		}
	}
}

func (p *Pool) terminal() {
	// show Pool detail
	for _, list := range p.BufferTaskList {
		for _, t := range list {
			fmt.Printf("[list] buffer task detail : %v %v, %v, %v \n", t.id, t.uuid, t.taskId, t.data)
		}
	}
}

//检测任务状态
func (p *Pool) Supervisor(appName string) {
	var duration = 1800
	//check
	p.check(appName, duration)
	//poll
	var t = time.Tick(1800 * time.Second)
	ch, err := GetChannel(appName)
	if err != nil {
		return
	}
	for {
		select {
		case <-t:
			p.check(appName, duration)
		case <-ch.ProcessorStop:
			return
		}
	}
}

func (p *Pool) check(appName string, duration int) {
	var db = mysql.GetInstance(false)
	var taskStatus = []int{TaskStatusFail, TaskStatusReady, TaskStatusStart, TaskStatusUnknown}
	var currentUnixTimeStamp = int(time.Now().Unix())
	var timePointEnd = helper.FormatTime(helper.TimeFormatYmdHis, currentUnixTimeStamp-duration)
	var timePointStart = helper.FormatTime(helper.TimeFormatYmdHis, currentUnixTimeStamp-2*duration)
	var taskListRecord = db.Find(records.RecordNameCensusTaskInfo).Select("id", "uuid", "action", "data", "status", "explain").
		Where("app_name", "=", appName).AndWhere("status", "in", taskStatus).
		AndWhere("create_time", ">", timePointStart).AndWhere("create_time", "<", timePointEnd).
		Execute().FetchAll()
	if taskListRecord.Len() == 0 {
		return
	}
	var msgList = make([]string, 0)
	uuid := utils.MakeUUID()
	for _, o := range taskListRecord.AllRecord() {
		r := o.(*records.CensusTaskInfo)
		msg := fmt.Sprintf("taskPriId:(%v);taskUUId:(%v);appName:(%v);action:(%v);status:(%v);data:(%v);explain:(%v)\n",
			r.Id, r.Uuid, appName, r.Action, taskStatusNote[r.Status], r.Data, r.Explain)
		msgList = append(msgList, msg)
		logger.Warning("task-problem", uuid, "detail problem", msg)
	}

}
